package com.feng.cloud.common.customer;

import com.alibaba.fastjson.JSON;
import com.feng.cloud.common.constants.RocketMqConstant;
import com.feng.cloud.common.message.DelayMessage;
import com.feng.cloud.common.utils.DelayLevelCalculate;
import lombok.SneakyThrows;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.Duration;
import java.time.Instant;
import java.util.Date;

/**
 * @ClassName DefaultDelayCustomer
 * @Description TODO
 * @Author Better Feng
 * @Date 2021/1/25 16:24
 **/
@RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.group}",topic = RocketMqConstant.DEFAULT_DELAY_TOPIC)
public class DefaultDelayCustomer implements RocketMQListener<MessageExt> {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @SneakyThrows
    @Override
    public void onMessage(MessageExt messageExt) {
        String message = new String(messageExt.getBody());
        DelayMessage delayMessage = JSON.parseObject(message, DelayMessage.class);
        Date startDeliverDate = delayMessage.getStartDeliverDate();
        long duration = Duration.between(Instant.now(), startDeliverDate.toInstant()).getSeconds();
        int level = DelayLevelCalculate.calculateDefault(duration);

        Message msg = new Message(delayMessage.getTopic(), delayMessage.getTags(), delayMessage.getBody());
        if (level > RocketMqConstant.MINI_DELAY_LEVEL) {
            // 如果大于最小延时等级，设置延时发送
            msg.setDelayTimeLevel(level);
            msg.setTopic(RocketMqConstant.DEFAULT_DELAY_TOPIC);
            msg.setBody(messageExt.getBody());
        }
        rocketMQTemplate.getProducer().send(msg);
    }
}
